Personal Computer World 2009 February/PCWFEB09.iso/Software/Resources/Chat & Communication/Digsby build 37/digsby_setup.exe/lib/dns/rdataset.pyo

Python Compiled Bytecode | 2008-10-13 | 6.5 KB | 217 lines

# Source Generated with Decompyle++
# File: in.pyo (Python 2.5)

import random
import StringIO
import struct
import dns.exception as dns
import dns.rdatatype as dns
import dns.rdataclass as dns
import dns.rdata as dns
import dns.set as dns
SimpleSet = dns.set.Set

class DifferingCovers(dns.exception.DNSException):
    pass


class IncompatibleTypes(dns.exception.DNSException):
    pass


class Rdataset(dns.set.Set):
    __slots__ = [
        'rdclass',
        'rdtype',
        'covers',
        'ttl']

    def __init__(self, rdclass, rdtype, covers = dns.rdatatype.NONE):
        super(Rdataset, self).__init__()
        self.rdclass = rdclass
        self.rdtype = rdtype
        self.covers = covers
        self.ttl = 0


    def _clone(self):
        obj = super(Rdataset, self)._clone()
        obj.rdclass = self.rdclass
        obj.rdtype = self.rdtype
        obj.covers = self.covers
        obj.ttl = self.ttl
        return obj


    def update_ttl(self, ttl):
        if len(self) == 0:
            self.ttl = ttl
        elif ttl < self.ttl:
            self.ttl = ttl



    def add(self, rd, ttl = None):
        if self.rdclass != rd.rdclass or self.rdtype != rd.rdtype:
            raise IncompatibleTypes

        if ttl is not None:
            self.update_ttl(ttl)

        if self.rdtype == dns.rdatatype.RRSIG or self.rdtype == dns.rdatatype.SIG:
            covers = rd.covers()
            if len(self) == 0 and self.covers == dns.rdatatype.NONE:
                self.covers = covers
            elif self.covers != covers:
                raise DifferingCovers


        if dns.rdatatype.is_singleton(rd.rdtype) and len(self) > 0:
            self.clear()

        super(Rdataset, self).add(rd)


    def union_update(self, other):
        self.update_ttl(other.ttl)
        super(Rdataset, self).union_update(other)


    def intersection_update(self, other):
        self.update_ttl(other.ttl)
        super(Rdataset, self).intersection_update(other)


    def update(self, other):
        self.update_ttl(other.ttl)
        super(Rdataset, self).update(other)


    def __repr__(self):
        if self.covers == 0:
            ctext = ''
        else:
            ctext = '(' + dns.rdatatype.to_text(self.covers) + ')'
        return '<DNS ' + dns.rdataclass.to_text(self.rdclass) + ' ' + dns.rdatatype.to_text(self.rdtype) + ctext + ' rdataset>'


    def __str__(self):
        return self.to_text()


    def __eq__(self, other):
        if not isinstance(other, Rdataset):
            return False

        if self.rdclass != other.rdclass and self.rdtype != other.rdtype or self.covers != other.covers:
            return False

        return super(Rdataset, self).__eq__(other)


    def __ne__(self, other):
        return not self.__eq__(other)


    def to_text(self, name = None, origin = None, relativize = True, override_rdclass = None, **kw):
        if name is not None:
            name = name.choose_relativity(origin, relativize)
            ntext = str(name)
            pad = ' '
        else:
            ntext = ''
            pad = ''
        s = StringIO.StringIO()
        if override_rdclass is not None:
            rdclass = override_rdclass
        else:
            rdclass = self.rdclass
        if len(self) == 0:
            print >>s, '%s%s%s %s' % (ntext, pad, dns.rdataclass.to_text(rdclass), dns.rdatatype.to_text(self.rdtype))
        else:
            for rd in self:
                print >>s, '%s%s%d %s %s %s' % (ntext, pad, self.ttl, dns.rdataclass.to_text(rdclass), dns.rdatatype.to_text(self.rdtype), rd.to_text(origin = origin, relativize = relativize, **kw))

        return s.getvalue()[:-1]


    def to_wire(self, name, file, compress = None, origin = None, override_rdclass = None, want_shuffle = True):
        if override_rdclass is not None:
            rdclass = override_rdclass
            want_shuffle = False
        else:
            rdclass = self.rdclass
        file.seek(0, 2)
        if len(self) == 0:
            name.to_wire(file, compress, origin)
            stuff = struct.pack('!HHIH', self.rdtype, rdclass, 0, 0)
            file.write(stuff)
            return 1
        elif want_shuffle:
            l = list(self)
            random.shuffle(l)
        else:
            l = self
        for rd in l:
            name.to_wire(file, compress, origin)
            stuff = struct.pack('!HHIH', self.rdtype, rdclass, self.ttl, 0)
            file.write(stuff)
            start = file.tell()
            rd.to_wire(file, compress, origin)
            end = file.tell()
            file.seek(start - 2)
            stuff = struct.pack('!H', end - start)
            file.write(stuff)
            file.seek(0, 2)

        return len(self)


    def match(self, rdclass, rdtype, covers):
        if self.rdclass == rdclass and self.rdtype == rdtype and self.covers == covers:
            return True

        return False



def from_text_list(rdclass, rdtype, ttl, text_rdatas):
    if isinstance(rdclass, str):
        rdclass = dns.rdataclass.from_text(rdclass)

    if isinstance(rdtype, str):
        rdtype = dns.rdatatype.from_text(rdtype)

    r = Rdataset(rdclass, rdtype)
    r.update_ttl(ttl)
    for t in text_rdatas:
        rd = dns.rdata.from_text(r.rdclass, r.rdtype, t)
        r.add(rd)

    return r


def from_text(rdclass, rdtype, ttl, *text_rdatas):
    return from_text_list(rdclass, rdtype, ttl, text_rdatas)


def from_rdata_list(ttl, rdatas):
    if len(rdatas) == 0:
        raise ValueError, 'rdata list must not be empty'

    r = None
    for rd in rdatas:
        if r is None:
            r = Rdataset(rd.rdclass, rd.rdtype)
            r.update_ttl(ttl)
            first_time = False

        r.add(rd)

    return r


def from_rdata(ttl, *rdatas):
    return from_rdata_list(ttl, rdatas)